package com.zyx.flink.realtime.func;

import com.alibaba.fastjson.JSONObject;
import com.zyx.flink.realtime.utils.DimUtil;
import com.zyx.flink.realtime.utils.ThreadPoolUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * @author zyx
 * @since 2021/6/11 20:16
 */
@Slf4j
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {

    private ExecutorService executorService;

    private final String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化线程池对象
        System.out.println("初始化线程池对象");
        executorService = ThreadPoolUtil.getInstance();
    }

    /**
     * 发送异步请求的方法
     *
     * @param obj          流中的事实数据
     * @param resultFuture 异步处理结束之后，返回结果
     */
    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) {
        executorService.submit(
                () -> {
                    try {
                        //发送异步请求
                        long start = System.currentTimeMillis();
                        //从流中事实数据获取key
                        String key = getKey(obj);
                        //根据维度的主键到维度表中进行查询
                        JSONObject dimInfoJsonObj = DimUtil.getDimInfoNoCache(tableName, Tuple2.of("id", key));
                        //System.out.println("维度数据Json格式：" + dimInfoJsonObj);
                        if (dimInfoJsonObj != null) {
                            //维度关联  流中的事实数据和查询出来的维度数据进行关联
                            join(obj, dimInfoJsonObj);
                        }
                        long end = System.currentTimeMillis();
                        log.info("++++++++++++++++++++异步维度查询耗时: {} ms", (end - start));
                        resultFuture.complete(Collections.singletonList(obj));
                    } catch (Exception e) {
                        e.printStackTrace();
                        throw new RuntimeException(tableName + "维度异步查询失败");
                    }
                }
        );
    }
}
